package com.gitee.huxing.courese.core;

import lombok.SneakyThrows;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @Description:
 * @ProjectName: JavaCourse
 * @Package: com.gitee.huxing.courese.core
 * @ClassName: Kmq
 * @Author: huxing
 * @DateTime: 2021-11-07 3:58 下午
 */
public final class Kmq {

    /** 订阅主题 **/
    private String topic;

    /** 区间 **/
    private int capactiy;

    /** 消息队列 **/
//    private LinkedBlockingQueue<KmqMessage> queue;
    private CustomMessageQueue<KmqMessage> queue;

    /** 消费者队列 **/
    private ConcurrentHashMap<String, Object> consumeQueue = new ConcurrentHashMap<>();

    public Kmq(String topic, int capactiy) {
        this.topic = topic;
        this.capactiy = capactiy;
        this.queue = new CustomMessageQueue<>(capactiy);
    }

    public boolean send(KmqMessage message){
        return queue.offer(message);
    }

    public KmqMessage poll(int consumerId){
        return queue.poll(consumerId);
    }

    @SneakyThrows
    public KmqMessage poll(int consumerId, long timeout){
        return queue.poll(consumerId, timeout, TimeUnit.MILLISECONDS);
    }
}
